package Market;

import java.util.Properties;
import java.util.concurrent.CountDownLatch;

import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.common.utils.Bytes;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.Topology;
import org.apache.kafka.streams.kstream.KGroupedStream;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.KTable;
import org.apache.kafka.streams.kstream.Materialized;
import org.apache.kafka.streams.state.KeyValueStore;

import Market.exchange.MarketData;
import Market.exchange.MarketDataSerDes;;

public class StreamComputerNoWindow {
    public static void main(String[] args) throws Exception {
        Properties props = new Properties();
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "streams-pipe");
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, MarketDataSerDes.class);

        final StreamsBuilder builder = new StreamsBuilder();

        // 使用一个ktable存储key,平均价格,价格波动
        KStream<String, MarketData> inputStream = builder.stream("stream-in");
        // inputStream.print(Printed.toSysOut());
        // inputStream.foreach((k,v)->System.out.print("k"));
        // 根据证券代码进行分类
        // KGroupedStream<Double,MarketData> kgroupStream =
        // inputStream.groupBy((key,value)->value.getChangeOver());
        KGroupedStream<String, MarketData> kgroupStream = inputStream.groupByKey();


        KTable<String, QuotaData> ktableMean = kgroupStream.aggregate(
                () -> new QuotaData(),
                (k, v, qd) -> {
                    qd.add(v.getLast());
                    return qd;
                },
                Materialized.<String, QuotaData, KeyValueStore<Bytes, byte[]>>as("store")
                        .withValueSerde(new QuotaDataSerDes()));

        // ktableMean.suppress(Suppressed.untilWindowCloses(Suppressed.BufferConfig.unbounded())).toStream().foreach((k, v) -> System.out.print(k.toString() + ": " + v.toString()));
        ktableMean.toStream().foreach((k, v) -> System.out.print(k.toString() + ": " + v.toString()));
        // ktableMean.toStream().to("stream-out");/* 将ktable中的数据变更变成流，具体方式有修改就产生一个新流 */

        final Topology topology = builder.build();
        final KafkaStreams streams = new KafkaStreams(topology, props);
        final CountDownLatch latch = new CountDownLatch(1);

        // attach shutdown handler to catch control-c
        Runtime.getRuntime().addShutdownHook(new Thread("streams-shutdown-hook") {
            @Override
            public void run() {
                streams.close();
                latch.countDown();
            }
        });

        try {
            streams.start();
            latch.await();
        } catch (Throwable e) {
            System.exit(1);
        }
        System.exit(0);
    }
}
